package com.wxy.demo.kafka;

import cn.hutool.json.JSONUtil;
import com.wxy.demo.dto.StepDto;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.SaslConfigs;

import java.util.ArrayList;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 *
 * </p>
 *
 * @author e-Xiaoyuan.Wang
 * @since 2022/11/7 18:13
 */
@Slf4j
public class ProducerDemo {
    public static void main(String[] args) throws InterruptedException {
        // kafka参数配置
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.168.40.128:9092,10.168.40.129:9092,10.168.40.130:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 添加认证
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"VMqjTk3hbcUSoL3h\";");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        // 构建数据
        ArrayList<StepDto> list = new ArrayList<>();
        for (int i = 0; i < 2; i++) {
            String orderNo = "No." + i;
            for (int j = 0; j < 10; j++) {
                list.add(new StepDto().setOrderNo(orderNo).setSort(j));
            }
        }
        // 推送消息
        for (StepDto stepDto : list) {
            String topic = "topic1";
            String content = JSONUtil.toJsonStr(stepDto);
            int size = producer.partitionsFor(topic).size();
            int partition = (stepDto.getOrderNo().hashCode() & Integer.MAX_VALUE) % size;
            producer.send(new ProducerRecord<>(topic, partition, null, content));
            log.info("发送消息：partition = {}, topic = {}, content = {}", partition, topic, content);
        }
        TimeUnit.SECONDS.sleep(1);
        producer.close();
    }
}
